/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.util.Collection;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;

import org.singularity.mapred.ReduceElement;
import org.singularity.mapred.VDME;

/**
 * @author vjache
 *
 */
public class DefaultReduceCache<KT> implements ReduceCache<KT> {

	private final Collection<Object> _mprIds;
	private final HashMap<Object, ConcurrentHashMap<KT, ConcurrentSkipListSet<Object>>> _mprCaches;
	private final ReduceElement<KT,?> _redElem;

	/**
	 * 
	 */
	public DefaultReduceCache(ReduceElement<KT,?> aRedElem, Collection<Object> aMapperIds) {
		_redElem = aRedElem;
		_mprIds = aMapperIds;
		_mprCaches = new HashMap<Object, ConcurrentHashMap<KT,ConcurrentSkipListSet<Object>>>(aMapperIds.size());
		
		for(Object mprId:aMapperIds)
		{
			_mprCaches.put(mprId, new ConcurrentHashMap<KT, ConcurrentSkipListSet<Object>>());
		}
		
	}

	/* (non-Javadoc)
	 * @see org.singularity.mapred.framework.ReduceCache#get(java.lang.Object, java.lang.Object)
	 */
	@Override
	public Iterable<?> get(Object aMapperId, KT aKey) {
		ConcurrentSkipListSet<Object> values = getValues(aMapperId, aKey);
		return values;
	}

	private ConcurrentSkipListSet<Object> getValues(Object aMapperId, KT aKey) {
		final ConcurrentHashMap<KT, ConcurrentSkipListSet<Object>> mprCache = _mprCaches.get(aMapperId);
		ConcurrentSkipListSet<Object> values = mprCache.get(aKey);
		if(values==null)
		{
			synchronized(mprCache)
			{
				values = mprCache.get(aKey);
				if(values==null)
				{
					mprCache.put(aKey, values = new ConcurrentSkipListSet<Object>());
				}
			}
		}
		return values;
	}

	/* (non-Javadoc)
	 * @see org.singularity.mapred.framework.ReduceCache#modify(org.singularity.mapred.VDME)
	 */
	@Override
	public boolean modify(VDME<KT, ?> aVdme) {
		final Object aMapperId = aVdme.getAnnotation(MapperPipe.MAPPER_ID_ANNOT);
		if(aMapperId==null)
			throw new RuntimeException(String.format("Required system annotation '%s' absent in VDME %s.", MapperPipe.MAPPER_ID_ANNOT,aVdme));
		ConcurrentSkipListSet<Object> values = getValues(aMapperId, aVdme.getKey());
		switch(aVdme.getCode())
		{
		case Add: return values.add(aVdme.getValue());
		case Remove: return values.remove(aVdme.getValue());
		}
		return false;
	}
	
	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("{");
		sb.append(" ReducerCacheImpl:");
		sb.append(getClass().getSimpleName());
		sb.append(",");
		sb.append(" reducer:");
		sb.append(_redElem);
		sb.append(",");
		sb.append(" mapperIds:");
		sb.append(_mprIds);
		sb.append("}");
		return sb.toString();
	}

}
